package com.li.flink.app

import com.alibaba.fastjson.JSON
import com.li.flink.domain.Access
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

object OsUserCntV1 {
  def main(args: Array[String]): Unit = {
    val inputPath = "/Users/lijiacen/Downloads/access.json"
    val conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build()
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val textStream = env.readTextFile(inputPath)
    val mapStream: DataStream[Access] = textStream.map(m => {
      try {
        val obj: Access = JSON.parseObject(m, classOf[Access])
        obj
      } catch {
        case ex: Exception => print("json数据解析失败：" + m)
          null
      }
    })
    val filterStream = mapStream.filter(m => {
      m.getEvent.equals("startup")
    }).map(m => {
      (m.getOs, m.getNu, 1)
    })
    val keySelector = new KeySelector[Tuple3[String, Int, Int], Tuple2[String, Int]] {
      override def getKey(in: Tuple3[String, Int, Int]): (String, Int) = {
        (in._1, in._2)
      }
    }
    //操作系统和新老用户维度
    val keyByStream = filterStream.keyBy(keySelector).sum(2)

    keyByStream.addSink(new RedisSink[(String, Int, Int)](conf, new RedisExampleMapper))

    env.execute("OsUserCntV1")
  }


}

class RedisExampleMapper extends RedisMapper[(String, Int, Int)] {
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET, "os-user_cnt")
  }

  override def getKeyFromData(data: (String, Int, Int)): String = data._1 + "_" + data._2

  override def getValueFromData(data: (String, Int, Int)): String = data._3 + ""
}